
/*
 * Copyright 2002-2022 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
/*
 *版权所有2002-2022原作者。
 *
 *根据Apache许可证2.0版许可（“许可证”）；
 *除非符合许可证的规定，否则您不得使用此文件。
 *您可以在以下网址获取许可证副本：
 *
 *https://www.apache.org/licenses/LICENSE-2.0
 *
 *除非适用法律要求或书面同意，软件
 *根据许可证进行的分发是在“按原样”的基础上进行的，
 *无任何明示或暗示的保证或条件。
 *有关管理权限的特定语言，请参阅许可证和
 *许可证下的限制。
 */

package org.springframework.scheduling.concurrent;

import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.springframework.core.task.AsyncListenableTaskExecutor;
import org.springframework.core.task.TaskDecorator;
import org.springframework.core.task.TaskRejectedException;
import org.springframework.lang.Nullable;
import org.springframework.scheduling.SchedulingTaskExecutor;
import org.springframework.util.Assert;
import org.springframework.util.ConcurrentReferenceHashMap;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureTask;

/**
 * JavaBean that allows for configuring a {@link java.util.concurrent.ThreadPoolExecutor}
 * in bean style (through its "corePoolSize", "maxPoolSize", "keepAliveSeconds", "queueCapacity"
 * properties) and exposing it as a Spring {@link org.springframework.core.task.TaskExecutor}.
 * This class is also well suited for management and monitoring (e.g. through JMX),
 * providing several useful attributes: "corePoolSize", "maxPoolSize", "keepAliveSeconds"
 * (all supporting updates at runtime); "poolSize", "activeCount" (for introspection only).
 *
 * <p>The default configuration is a core pool size of 1, with unlimited max pool size
 * and unlimited queue capacity. This is roughly equivalent to
 * {@link java.util.concurrent.Executors#newSingleThreadExecutor()}, sharing a single
 * thread for all tasks. Setting {@link #setQueueCapacity "queueCapacity"} to 0 mimics
 * {@link java.util.concurrent.Executors#newCachedThreadPool()}, with immediate scaling
 * of threads in the pool to a potentially very high number. Consider also setting a
 * {@link #setMaxPoolSize "maxPoolSize"} at that point, as well as possibly a higher
 * {@link #setCorePoolSize "corePoolSize"} (see also the
 * {@link #setAllowCoreThreadTimeOut "allowCoreThreadTimeOut"} mode of scaling).
 *
 * <p><b>NOTE:</b> This class implements Spring's
 * {@link org.springframework.core.task.TaskExecutor} interface as well as the
 * {@link java.util.concurrent.Executor} interface, with the former being the primary
 * interface, the other just serving as secondary convenience. For this reason, the
 * exception handling follows the TaskExecutor contract rather than the Executor contract,
 * in particular regarding the {@link org.springframework.core.task.TaskRejectedException}.
 *
 * <p>For an alternative, you may set up a ThreadPoolExecutor instance directly using
 * constructor injection, or use a factory method definition that points to the
 * {@link java.util.concurrent.Executors} class. To expose such a raw Executor as a
 * Spring {@link org.springframework.core.task.TaskExecutor}, simply wrap it with a
 * {@link org.springframework.scheduling.concurrent.ConcurrentTaskExecutor} adapter.
 *
 * @author Juergen Hoeller
 * @author Rémy Guihard
 * @author Sam Brannen
 * @since 2.0
 * @see org.springframework.core.task.TaskExecutor
 * @see java.util.concurrent.ThreadPoolExecutor
 * @see ThreadPoolExecutorFactoryBean
 * @see ConcurrentTaskExecutor
 */
/**
 *允许配置｛@link java.util.concurrent.ThreadPoolExecutor｝的JavaBean
 *在bean样式中（通过其“corePoolSize”、“maxPoolSize”，“keepAliveSeconds”、“queueCapacity”
 *属性），并将其公开为Spring｛@linkorg.springframework.core.task.TaskExecutor｝。
 *该类也非常适合于管理和监控（例如通过JMX），
 *提供了几个有用的属性：“corePoolSize”、“maxPoolSize”和“keepAliveSeconds”
 *（所有支持运行时更新）；“poolSize”，“activeCount”（仅用于内省）。
 *
 *<p>默认配置是核心池大小为1，最大池大小不受限制
 *以及无限制的队列容量。这大致相当于
 *｛@link java.util.concurrent.Executors#newSingleThreadExecutor（）｝，共享单个
 *所有任务的线程。将｛@link#setQueueCapacity“queueCapacity”｝设置为0模拟
 *｛@link java.util.concurrent.Executors#newCachedThreadPool（）｝，具有立即缩放功能
 *池中线程的数量增加到可能非常高的数量。还可以考虑设置
 *｛@link#setMaxPoolSize“maxPoolSize”｝在该点，以及可能更高的
 *｛@link#setCorePoolSize“corePoolSize”｝（另请参阅
 *｛@link#setAllowCoreThreadTimeOut“allowCoreThreadsTimeOut”｝缩放模式）。
 *
 *<p><b>注意：</b>这个类实现了Spring的
 *｛@link org.springframework.core.task.TaskExecutor｝接口以及
 *{@link java.util.concurrent.Executor}接口，前者是主要的
 *接口，另一个只是作为次要的便利。因此
 *异常处理遵循TaskExecutor合约而不是Executitor合约，
 *特别是关于{@linkorg.springframework.core.task.TaskRejectedException}。
 *
 *＜p＞作为替代方案，您可以使用
 *构造函数注入，或者使用指向
 *｛@link java.util.concurrent.Executors｝类。将这样一个原始的执行器公开为
 *Spring｛@linkorg.springframework.core.task.TaskExecutor｝，只需用
 *｛@link org.springframework.scheducing.concurrentTaskExecutitor｝适配器。
 *
 *@作者于尔根·霍勒
 *@作者雷米·吉哈德
 *@作者Sam Brannen
 *@自2.0起
 *@参见org.springframework.core.task.TaskExecutitor
 *@请参阅java.util.concurrent.ThreadPoolExecutor
 *@see线程池执行器FactoryBean
 *@请参阅并发任务执行器
 */
@SuppressWarnings({"serial", "deprecation"})
public class ThreadPoolTaskExecutor extends ExecutorConfigurationSupport
		implements AsyncListenableTaskExecutor, SchedulingTaskExecutor {

	private final Object poolSizeMonitor = new Object();

	private int corePoolSize = 1;

	private int maxPoolSize = Integer.MAX_VALUE;

	private int keepAliveSeconds = 60;

	private int queueCapacity = Integer.MAX_VALUE;

	private boolean allowCoreThreadTimeOut = false;

	private boolean prestartAllCoreThreads = false;

	@Nullable
	private TaskDecorator taskDecorator;

	@Nullable
	private ThreadPoolExecutor threadPoolExecutor;

	// Runnable decorator to user-level FutureTask, if different
	// 可运行的装饰器到用户级FutureTask，如果不同
	private final Map<Runnable, Object> decoratedTaskMap =
			new ConcurrentReferenceHashMap<>(16, ConcurrentReferenceHashMap.ReferenceType.WEAK);


	/**
	 * Set the ThreadPoolExecutor's core pool size.
	 * Default is 1.
	 * <p><b>This setting can be modified at runtime, for example through JMX.</b>
	 */
	/**
	 *设置ThreadPoolExecutitor的核心池大小。
	 *默认值为1。
	 *<p><b>可以在运行时修改此设置，例如通过JMX</b>
	 */
	public void setCorePoolSize(int corePoolSize) {
		synchronized (this.poolSizeMonitor) {
			if (this.threadPoolExecutor != null) {
				this.threadPoolExecutor.setCorePoolSize(corePoolSize);
			}
			this.corePoolSize = corePoolSize;
		}
	}

	/**
	 * Return the ThreadPoolExecutor's core pool size.
	 */
	/**
	 *返回ThreadPoolExecutitor的核心池大小。
	 */
	public int getCorePoolSize() {
		synchronized (this.poolSizeMonitor) {
			return this.corePoolSize;
		}
	}

	/**
	 * Set the ThreadPoolExecutor's maximum pool size.
	 * Default is {@code Integer.MAX_VALUE}.
	 * <p><b>This setting can be modified at runtime, for example through JMX.</b>
	 */
	/**
	 *设置ThreadPoolExecutitor的最大池大小。
	 *默认值为｛@code Integer.MAX_VALUE｝。
	 *<p><b>可以在运行时修改此设置，例如通过JMX</b>
	 */
	public void setMaxPoolSize(int maxPoolSize) {
		synchronized (this.poolSizeMonitor) {
			if (this.threadPoolExecutor != null) {
				this.threadPoolExecutor.setMaximumPoolSize(maxPoolSize);
			}
			this.maxPoolSize = maxPoolSize;
		}
	}

	/**
	 * Return the ThreadPoolExecutor's maximum pool size.
	 */
	/**
	 *返回ThreadPoolExecutitor的最大池大小。
	 */
	public int getMaxPoolSize() {
		synchronized (this.poolSizeMonitor) {
			return this.maxPoolSize;
		}
	}

	/**
	 * Set the ThreadPoolExecutor's keep-alive seconds.
	 * <p>Default is 60.
	 * <p><b>This setting can be modified at runtime, for example through JMX.</b>
	 */
	/**
	 *设置ThreadPoolExecutor的保持活动秒数。
	 *<p>默认值为60。
	 *<p><b>可以在运行时修改此设置，例如通过JMX</b>
	 */
	public void setKeepAliveSeconds(int keepAliveSeconds) {
		synchronized (this.poolSizeMonitor) {
			if (this.threadPoolExecutor != null) {
				this.threadPoolExecutor.setKeepAliveTime(keepAliveSeconds, TimeUnit.SECONDS);
			}
			this.keepAliveSeconds = keepAliveSeconds;
		}
	}

	/**
	 * Return the ThreadPoolExecutor's keep-alive seconds.
	 */
	/**
	 *返回ThreadPoolExecutitor的保持活动秒数。
	 */
	public int getKeepAliveSeconds() {
		synchronized (this.poolSizeMonitor) {
			return this.keepAliveSeconds;
		}
	}

	/**
	 * Set the capacity for the ThreadPoolExecutor's BlockingQueue.
	 * <p>Default is {@code Integer.MAX_VALUE}.
	 * <p>Any positive value will lead to a LinkedBlockingQueue instance;
	 * any other value will lead to a SynchronousQueue instance.
	 * @see java.util.concurrent.LinkedBlockingQueue
	 * @see java.util.concurrent.SynchronousQueue
	 */
	/**
	 *设置ThreadPoolExecutitor的BlockingQueue的容量。
	 *<p>默认值为｛@code Integer.MAX_VALUE｝。
	 *<p>任何正值都将导致LinkedBlockingQueue实例；
	 *任何其他值都将导致SynchronousQueue实例。
	 *@参见java.util.concurrent.LinkedBlockingQueue
	 *@参见java.util.courrent.SynchroniousQueue
	 */
	public void setQueueCapacity(int queueCapacity) {
		this.queueCapacity = queueCapacity;
	}

	/**
	 * Return the capacity for the ThreadPoolExecutor's BlockingQueue.
	 * @since 5.3.21
	 * @see #setQueueCapacity(int)
	 */
	/**
	 *返回ThreadPoolExecutitor的BlockingQueue的容量。
	 *@自21年3月3日起
	 *@参见#setQueueCapacity（int）
	 */
	public int getQueueCapacity() {
		return this.queueCapacity;
	}

	/**
	 * Specify whether to allow core threads to time out. This enables dynamic
	 * growing and shrinking even in combination with a non-zero queue (since
	 * the max pool size will only grow once the queue is full).
	 * <p>Default is "false".
	 * @see java.util.concurrent.ThreadPoolExecutor#allowCoreThreadTimeOut(boolean)
	 */
	/**
	 *指定是否允许核心线程超时。这实现了动态
	 *增长和收缩，即使与非零队列结合使用（因为
	 *最大池大小将仅在队列已满时增长）。
	 *<p>默认值为“false”。
	 *@请参阅java.util.concurrent.ThreadPoolExecutor#allowCoreThreadTimeOut（布尔值）
	 */
	public void setAllowCoreThreadTimeOut(boolean allowCoreThreadTimeOut) {
		this.allowCoreThreadTimeOut = allowCoreThreadTimeOut;
	}

	/**
	 * Specify whether to start all core threads, causing them to idly wait for work.
	 * <p>Default is "false".
	 * @since 5.3.14
	 * @see java.util.concurrent.ThreadPoolExecutor#prestartAllCoreThreads
	 */
	/**
	 *指定是否启动所有核心线程，使它们空闲等待工作。
	 *<p>默认值为“false”。
	 *@自2014年5月14日起
	 *@请参阅java.util.concurrent.ThreadPoolExecutor#prestartAllCoreThreads
	 */
	public void setPrestartAllCoreThreads(boolean prestartAllCoreThreads) {
		this.prestartAllCoreThreads = prestartAllCoreThreads;
	}

	/**
	 * Specify a custom {@link TaskDecorator} to be applied to any {@link Runnable}
	 * about to be executed.
	 * <p>Note that such a decorator is not necessarily being applied to the
	 * user-supplied {@code Runnable}/{@code Callable} but rather to the actual
	 * execution callback (which may be a wrapper around the user-supplied task).
	 * <p>The primary use case is to set some execution context around the task's
	 * invocation, or to provide some monitoring/statistics for task execution.
	 * <p><b>NOTE:</b> Exception handling in {@code TaskDecorator} implementations
	 * is limited to plain {@code Runnable} execution via {@code execute} calls.
	 * In case of {@code #submit} calls, the exposed {@code Runnable} will be a
	 * {@code FutureTask} which does not propagate any exceptions; you might
	 * have to cast it and call {@code Future#get} to evaluate exceptions.
	 * See the {@code ThreadPoolExecutor#afterExecute} javadoc for an example
	 * of how to access exceptions in such a {@code Future} case.
	 * @since 4.3
	 */
	/**
	 *指定要应用于任何｛@link Runnable｝的自定义｛@linkTaskDecorator｝
	 *即将执行。
	 *<p>请注意，这样的装饰器不一定要应用于
	 *用户提供的｛@code Runnable｝/｛@codeCallable｝，而不是实际的
	 *执行回调（可以是用户提供的任务的包装器）。
	 *<p>主要用例是围绕任务的
	 *调用，或者为任务执行提供一些监控/统计信息。
	 *<p><b>注意：</b>｛@code TaskDecorator｝实现中的异常处理
	 *仅限于通过｛@code execute｝调用执行普通的｛@codeRunnable｝。
	 *在｛@code#submit｝调用的情况下，公开的｛@codeRunnable｝将是
	 *｛@code FutureTask｝，它不传播任何异常；你可能
	 *必须强制转换它并调用｛@code Future#get｝来计算异常。
	 *有关示例，请参阅{@code ThreadPoolExecutor#afterExecute}javadoc
	 *关于如何在这种{@codeFuture}情况下访问异常。
	 *@自4.3起
	 */
	public void setTaskDecorator(TaskDecorator taskDecorator) {
		this.taskDecorator = taskDecorator;
	}


	/**
	 * Note: This method exposes an {@link ExecutorService} to its base class
	 * but stores the actual {@link ThreadPoolExecutor} handle internally.
	 * Do not override this method for replacing the executor, rather just for
	 * decorating its {@code ExecutorService} handle or storing custom state.
	 */
	/**
	 *注意：此方法向其基类公开｛@link ExecutiorService｝
	 *但在内部存储实际的{@link ThreadPoolExecution}句柄。
	 *不要重写此方法来替换执行器，而仅用于
	 *装饰其{@codeExecutiorService}句柄或存储自定义状态。
	 */
	@Override
	protected ExecutorService initializeExecutor(
			ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {

		BlockingQueue<Runnable> queue = createQueue(this.queueCapacity);

		ThreadPoolExecutor executor;
		if (this.taskDecorator != null) {
			executor = new ThreadPoolExecutor(
					this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS,
					queue, threadFactory, rejectedExecutionHandler) {
				@Override
				public void execute(Runnable command) {
					Runnable decorated = taskDecorator.decorate(command);
					if (decorated != command) {
						decoratedTaskMap.put(decorated, command);
					}
					super.execute(decorated);
				}
			};
		}
		else {
			executor = new ThreadPoolExecutor(
					this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS,
					queue, threadFactory, rejectedExecutionHandler);

		}

		if (this.allowCoreThreadTimeOut) {
			executor.allowCoreThreadTimeOut(true);
		}
		if (this.prestartAllCoreThreads) {
			executor.prestartAllCoreThreads();
		}

		this.threadPoolExecutor = executor;
		return executor;
	}

	/**
	 * Create the BlockingQueue to use for the ThreadPoolExecutor.
	 * <p>A LinkedBlockingQueue instance will be created for a positive
	 * capacity value; a SynchronousQueue else.
	 * @param queueCapacity the specified queue capacity
	 * @return the BlockingQueue instance
	 * @see java.util.concurrent.LinkedBlockingQueue
	 * @see java.util.concurrent.SynchronousQueue
	 */
	/**
	 *创建用于ThreadPoolExecutor的BlockingQueue。
	 *<p>将为阳性创建LinkedBlockingQueue实例
	 *容量值；a SynchronousQueue else。
	 *@param queueCapacity指定的队列容量
	 *@return BlockingQueue实例
	 *@参见java.util.concurrent.LinkedBlockingQueue
	 *@参见java.util.courrent.SynchroniousQueue
	 */
	protected BlockingQueue<Runnable> createQueue(int queueCapacity) {
		if (queueCapacity > 0) {
			return new LinkedBlockingQueue<>(queueCapacity);
		}
		else {
			return new SynchronousQueue<>();
		}
	}

	/**
	 * Return the underlying ThreadPoolExecutor for native access.
	 * @return the underlying ThreadPoolExecutor (never {@code null})
	 * @throws IllegalStateException if the ThreadPoolTaskExecutor hasn't been initialized yet
	 */
	/**
	 *返回用于本机访问的基础ThreadPoolExecutor。
	 *@return底层ThreadPoolExecutor（从不｛@code null｝）
	 *如果ThreadPoolTaskExecutor尚未初始化，则@throws IllegalStateException
	 */
	public ThreadPoolExecutor getThreadPoolExecutor() throws IllegalStateException {
		Assert.state(this.threadPoolExecutor != null, "ThreadPoolTaskExecutor not initialized");
		return this.threadPoolExecutor;
	}

	/**
	 * Return the current pool size.
	 * @see java.util.concurrent.ThreadPoolExecutor#getPoolSize()
	 */
	/**
	 *返回当前池大小。
	 *@请参阅java.util.concurrent.ThreadPoolExecutor#getPoolSize（）
	 */
	public int getPoolSize() {
		if (this.threadPoolExecutor == null) {
			// Not initialized yet: assume core pool size.
			// 尚未初始化：假设核心池大小。
			return this.corePoolSize;
		}
		return this.threadPoolExecutor.getPoolSize();
	}

	/**
	 * Return the current queue size.
	 * @since 5.3.21
	 * @see java.util.concurrent.ThreadPoolExecutor#getQueue()
	 */
	/**
	 *返回当前队列大小。
	 *@自21年3月3日起
	 *@请参阅java.util.concurrent.ThreadPoolExecutor#getQueue（）
	 */
	public int getQueueSize() {
		if (this.threadPoolExecutor == null) {
			// Not initialized yet: assume no queued tasks.
			// 尚未初始化：假设没有排队的任务。
			return 0;
		}
		return this.threadPoolExecutor.getQueue().size();
	}

	/**
	 * Return the number of currently active threads.
	 * @see java.util.concurrent.ThreadPoolExecutor#getActiveCount()
	 */
	/**
	 *返回当前活动线程的数量。
	 *@请参阅java.util.concurrent.ThreadPoolExecutor#getActiveCount（）
	 */
	public int getActiveCount() {
		if (this.threadPoolExecutor == null) {
			// Not initialized yet: assume no active threads.
			// 尚未初始化：假设没有活动线程。
			return 0;
		}
		return this.threadPoolExecutor.getActiveCount();
	}


	@Override
	public void execute(Runnable task) {
		Executor executor = getThreadPoolExecutor();
		try {
			executor.execute(task);
		}
		catch (RejectedExecutionException ex) {
			throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
		}
	}

	@Deprecated
	@Override
	public void execute(Runnable task, long startTimeout) {
		execute(task);
	}

	@Override
	public Future<?> submit(Runnable task) {
		ExecutorService executor = getThreadPoolExecutor();
		try {
			return executor.submit(task);
		}
		catch (RejectedExecutionException ex) {
			throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
		}
	}

	@Override
	public <T> Future<T> submit(Callable<T> task) {
		ExecutorService executor = getThreadPoolExecutor();
		try {
			return executor.submit(task);
		}
		catch (RejectedExecutionException ex) {
			throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
		}
	}

	@Override
	public ListenableFuture<?> submitListenable(Runnable task) {
		ExecutorService executor = getThreadPoolExecutor();
		try {
			ListenableFutureTask<Object> future = new ListenableFutureTask<>(task, null);
			executor.execute(future);
			return future;
		}
		catch (RejectedExecutionException ex) {
			throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
		}
	}

	@Override
	public <T> ListenableFuture<T> submitListenable(Callable<T> task) {
		ExecutorService executor = getThreadPoolExecutor();
		try {
			ListenableFutureTask<T> future = new ListenableFutureTask<>(task);
			executor.execute(future);
			return future;
		}
		catch (RejectedExecutionException ex) {
			throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
		}
	}

	@Override
	protected void cancelRemainingTask(Runnable task) {
		super.cancelRemainingTask(task);
		// Cancel associated user-level Future handle as well
		// 同时取消关联的用户级Future句柄
		Object original = this.decoratedTaskMap.get(task);
		if (original instanceof Future) {
			((Future<?>) original).cancel(true);
		}
	}

}
